package com.bkgy.component;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.bkgy.modular.datacollect_bx.bean.JDataBigToSmall;
import com.bkgy.modular.datacollect_bx.bean.JDataSave621;
import com.bkgy.modular.datacollect_bx.bean.JDataSave622;
import com.bkgy.modular.datacollect_bx.bean.JDataSave642;
import com.bkgy.modular.datacollect_bx.service.JDataSaveService621;
import com.bkgy.modular.datacollect_bx.service.JDataSaveService622;
import com.bkgy.modular.datacollect_bx.service.JDataSaveService642;
import com.bkgy.modular.tbbxfilecollect.TbBXFileCollect;
import com.bkgy.modular.tbbxfilecollect.TbBXFileCollectDao;
import com.bkgy.controller.tbbxrecord.dao.TbBxFileSaveDao;
import com.bkgy.controller.tbbxrecord.entity.TbBxFileSave;
import com.bkgy.util.SFtpUtils;
import com.jcraft.jsch.ChannelSftp;
import com.jcraft.jsch.JSchException;
import com.jcraft.jsch.SftpException;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.compress.compressors.gzip.GzipCompressorInputStream;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
import org.springframework.web.client.RestTemplate;

import javax.annotation.Resource;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

@Component
@Slf4j
public class ReReadBigFile {

    @Value("${sftp_host_621}")
    String host621;

    @Value("${sftp_host_622}")
    String host622;

    @Value("${sftp_host_642}")
    String host642;

    @Resource
    private SFtpUtils sFtpUtils;

    @Resource
    private TbBxFileSaveDao tbBxFileSaveDao;

    @Resource
    private TbBXFileCollectDao tbBXFileCollectDao;

    @Resource
    private JDataSaveService621 jDataSaveService621;

    @Resource
    private JDataSaveService622 jDataSaveService622;

    @Resource
    private JDataSaveService642 jDataSaveService642;

    @Resource
    private RestTemplate restTemplate;


    /**
     * 查出所有的为1的数据
     *
     * @return
     */
    private List<TbBxFileSave> findList() {
        QueryWrapper<TbBxFileSave> wrapper = new QueryWrapper<>();
        wrapper.lambda().eq(TbBxFileSave::getStatusftp, 1L);
        return tbBxFileSaveDao.selectList(wrapper);
    }

    /**
     * 处理大文件数据
     * @throws InterruptedException
     */

    public void calBigFile() throws InterruptedException {

        List<TbBxFileSave> list = findList();
        //遍历各个未处理的大文件消息
        for (TbBxFileSave tbBxFileSave : list) {

            QueryWrapper<TbBXFileCollect> wrapper = new QueryWrapper<>();
            //621处理
            if (tbBxFileSave.getLine().equals("C621")) {
                JDataSave621 jDataSave621 = new JDataSave621();
                //初始化基本信息
                JSONObject key = JSONObject.parseObject(tbBxFileSave.getMessage()).getJSONObject("key");
                jDataSave621.setExtid(key.getString("extid"));
                jDataSave621.setEntid(key.getString("entid"));
                jDataSave621.setStart(key.getString("start"));
                jDataSave621.setEnd(key.getString("end"));
                jDataSave621.setCreatetime(new Date());
                Map<String, List<Double>> map = new HashMap<>();
                //下面都是为了填满map而做的事
                wrapper.lambda().eq(TbBXFileCollect::getLine, "C621");
                //获取需要存储的曲线名称别名
                List<TbBXFileCollect> collects = tbBXFileCollectDao.selectList(wrapper);
                //将所需的信息按照文件名头分组，集中处理
                Map<Double, List<TbBXFileCollect>> mapValue = collects.stream().collect(Collectors.groupingBy(TbBXFileCollect::getFile_num));
                //k文件名头，v内容(将row里面的东西填满)
                mapValue.forEach((k, v) -> {
                    //文件编号
                    int i = k.intValue();
                    //文件路径
                    String directory = key.getString("path");
                    //String  directory="/data/dsc/C621"+directory1.substring(8,directory1.length());
                    //System.out.println(directory);
                    //文件名
                    String downloadFile = i + "-" + tbBxFileSave.getCoil_id() + ".json.tar";
                    //解析文件路径
                    ChannelSftp sftp = sFtpUtils.sftpConnect(host621);
                    if (directory != null && !"".equals(directory)) {
                        try {
                            sftp.cd(directory);
                        } catch (SftpException e) {
                            throw new RuntimeException(e);
                        }
                    }
                    //处理文件流
                    InputStream inputStream = null;
                    try {
                        inputStream = sftp.get(directory + downloadFile);
                    } catch (SftpException e) {
                        return;
                        //e.printStackTrace();
                    }
                    //判断文件大小（需要保证自己电脑有D://TEMP/目录，没有可以修改）
                    sFtpUtils.loadFile(inputStream, "D://TEMP/" + downloadFile);
                    File file = new File("D://TEMP/" + downloadFile);
                    if (file.length() > 1024 * 1024) {
                        log.error(directory + downloadFile + "大文件处理文件大小异常---" + host621);
                        try {
                            inputStream.close();
                        } catch (IOException e) {
                            throw new RuntimeException(e);
                        }
                        file.delete();
                        sftp.disconnect();
                        try {
                            sftp.getSession().disconnect();
                        } catch (JSchException e) {
                            throw new RuntimeException(e);
                        }
                        //如果小文件也是很大
                        tbBxFileSave.setStatusftp(8.0);
                        UpdateWrapper<TbBxFileSave> updateWrapper = new UpdateWrapper<>();
                        updateWrapper.lambda().eq(TbBxFileSave::getCoil_id, tbBxFileSave.getCoil_id());
                        tbBxFileSaveDao.update(tbBxFileSave, updateWrapper);
                        return;
                    }
                    file.delete();
                    //读完再读一次
                    InputStream inputStream1 = null;
                    try {
                        inputStream1 = sftp.get(directory + downloadFile);
                    } catch (SftpException e) {
                        throw new RuntimeException(e);
                    }

                    if (inputStream1 != null) {
                        try {
                            //开始处理大文件中的小文件
                            GzipCompressorInputStream gzip = null;
                            gzip = new GzipCompressorInputStream(inputStream);
                            String s = sFtpUtils.readFileToString(gzip);
                            if (StringUtils.hasText(s)) {
                                //解析出宝信原始数据
                                JDataBigToSmall jDataBigToSmall = JSON.parseObject(s, JDataBigToSmall.class);
                                List<List<String>> row = jDataBigToSmall.getRow();
                                //循环获取里面的数据
                                for (TbBXFileCollect tbBXFileCollect : v) {
                                    int i1 = tbBXFileCollect.getRow_id().intValue();
                                    //根据指定的下标获取需要的值
                                    List<String> value = row.get(i1);
                                    List<Double> collect = value.stream().map(Double::parseDouble).collect(Collectors.toList());
                                    //存入别名，加内容
                                    map.put(tbBXFileCollect.getL1_variable_alias(), collect);
                                }
                            }
                            gzip.close();
                            inputStream1.close();
                            sftp.disconnect();
                            sftp.getSession().disconnect();
                        } catch (JSchException e) {
                            throw new RuntimeException(e);
                        } catch (IOException e) {
                            throw new RuntimeException(e);
                        }
                    }

                });
                //存入MongoDB
                jDataSave621.setRow(map);
                jDataSaveService621.insertData(jDataSave621);
                //如果处理完之后
                tbBxFileSave.setStatusftp(9.0);
                UpdateWrapper<TbBxFileSave> updateWrapper = new UpdateWrapper<>();
                updateWrapper.lambda().eq(TbBxFileSave::getCoil_id, tbBxFileSave.getCoil_id());
                tbBxFileSaveDao.update(tbBxFileSave, updateWrapper);
                JSONObject jsonObject = new JSONObject();
                jsonObject.put("volumeNo", key.getString("extid"));
                jsonObject.put("line", "C621");
                String jsonString = jsonObject.toJSONString();
                restTemplate.postForObject("http://170.0.200.112:1234/send_msg_to_models", jsonString, String.class);
                Thread.sleep(3000);
            }

            //622处理
            if (tbBxFileSave.getLine().equals("C622")) {
                JDataSave622 jDataSave622 = new JDataSave622();
                //初始化基本信息
                JSONObject key = JSONObject.parseObject(tbBxFileSave.getMessage()).getJSONObject("key");
                jDataSave622.setExtid(key.getString("extid"));
                jDataSave622.setEntid(key.getString("entid"));
                jDataSave622.setStart(key.getString("start"));
                jDataSave622.setEnd(key.getString("end"));
                jDataSave622.setCreatetime(new Date());
                Map<String, List<Double>> map = new HashMap<>();
                //下面都是为了填满map而做的事
                wrapper.lambda().eq(TbBXFileCollect::getLine, "C622");
                //获取需要存储的曲线名称别名
                List<TbBXFileCollect> collects = tbBXFileCollectDao.selectList(wrapper);
                //将所需的信息按照文件名头分组，集中处理
                Map<Double, List<TbBXFileCollect>> mapValue = collects.stream().collect(Collectors.groupingBy(TbBXFileCollect::getFile_num));
                //k文件名头，v内容(将row里面的东西填满)
                mapValue.forEach((k, v) -> {
                    //文件编号
                    int i = k.intValue();
                    //文件路径
                    String directory = key.getString("path");
                    //文件名
                    String downloadFile = i + "-" + tbBxFileSave.getCoil_id() + ".json.tar";
                    //解析文件路径
                    ChannelSftp sftp = sFtpUtils.sftpConnect(host622);
                    if (directory != null && !"".equals(directory)) {
                        try {
                            sftp.cd(directory);
                        } catch (SftpException e) {
                            throw new RuntimeException(e);
                        }
                    }
                    //处理文件流
                    InputStream inputStream = null;
                    try {
                        inputStream = sftp.get(directory + downloadFile);
                    } catch (SftpException e) {
                        return;
                       // throw new RuntimeException(e);
                    }

                    //判断文件大小（需要保证自己电脑有D://TEMP/目录，没有可以修改）
                    sFtpUtils.loadFile(inputStream, "D://TEMP/" + downloadFile);
                    File file = new File("D://TEMP/" + downloadFile);
                    if (file.length() > 1024 * 1024) {
                        log.error(directory + downloadFile + "大文件处理文件大小异常---" + host622);
                        try {
                            inputStream.close();
                        } catch (IOException e) {
                            throw new RuntimeException(e);
                        }
                        file.delete();
                        sftp.disconnect();
                        try {
                            sftp.getSession().disconnect();
                        } catch (JSchException e) {
                            throw new RuntimeException(e);
                        }
                        //如果小文件也是很大
                        tbBxFileSave.setStatusftp(8.0);
                        UpdateWrapper<TbBxFileSave> updateWrapper = new UpdateWrapper<>();
                        updateWrapper.lambda().eq(TbBxFileSave::getCoil_id, tbBxFileSave.getCoil_id());
                        tbBxFileSaveDao.update(tbBxFileSave, updateWrapper);
                        return;
                    }
                    file.delete();
                    //读完再读一次
                    InputStream inputStream1 = null;
                    try {
                        inputStream1 = sftp.get(directory + downloadFile);
                    } catch (SftpException e) {
                        throw new RuntimeException(e);
                    }

                    if (inputStream1 != null) {
                        try {
                            //开始处理大文件中的小文件
                            GzipCompressorInputStream gzip = null;
                            gzip = new GzipCompressorInputStream(inputStream);
                            String s = sFtpUtils.readFileToString(gzip);
                            if (StringUtils.hasText(s)) {
                                //解析出宝信原始数据
                                JDataBigToSmall jDataBigToSmall = JSON.parseObject(s, JDataBigToSmall.class);
                                List<List<String>> row = jDataBigToSmall.getRow();
                                //循环获取里面的数据
                                for (TbBXFileCollect tbBXFileCollect : v) {
                                    int i1 = tbBXFileCollect.getRow_id().intValue();
                                    //根据指定的下标获取需要的值
                                    List<String> value = row.get(i1);
                                    List<Double> collect = value.stream().map(Double::parseDouble).collect(Collectors.toList());
                                    //存入别名，加内容
                                    map.put(tbBXFileCollect.getL1_variable_alias(), collect);
                                }
                            }
                            gzip.close();
                            inputStream1.close();
                            sftp.disconnect();
                            sftp.getSession().disconnect();
                        } catch (JSchException e) {
                            throw new RuntimeException(e);
                        } catch (IOException e) {
                            throw new RuntimeException(e);
                        }
                    }

                });
                //存入MongoDB
                jDataSave622.setRow(map);
                jDataSaveService622.insertData(jDataSave622);
                //如果处理完之后
                tbBxFileSave.setStatusftp(9.0);
                UpdateWrapper<TbBxFileSave> updateWrapper = new UpdateWrapper<>();
                updateWrapper.lambda().eq(TbBxFileSave::getCoil_id, tbBxFileSave.getCoil_id());
                tbBxFileSaveDao.update(tbBxFileSave, updateWrapper);
                JSONObject jsonObject = new JSONObject();
                jsonObject.put("volumeNo", key.getString("extid"));
                jsonObject.put("line", "C622");
                String jsonString = jsonObject.toJSONString();
                restTemplate.postForObject("http://170.0.200.112:1234/send_msg_to_models", jsonString, String.class);
                Thread.sleep(3000);
            }

            //642处理
            if (tbBxFileSave.getLine().equals("C642")) {
                JDataSave642 jDataSave642 = new JDataSave642();
                //初始化基本信息
                JSONObject key = JSONObject.parseObject(tbBxFileSave.getMessage()).getJSONObject("key");
                jDataSave642.setExtid(key.getString("extid"));
                jDataSave642.setEntid(key.getString("entid"));
                jDataSave642.setStart(key.getString("start"));
                jDataSave642.setEnd(key.getString("end"));
                jDataSave642.setCreatetime(new Date());
                Map<String, List<Double>> map = new HashMap<>();
                //下面都是为了填满map而做的事
                wrapper.lambda().eq(TbBXFileCollect::getLine, "C642");
                //获取需要存储的曲线名称别名
                List<TbBXFileCollect> collects = tbBXFileCollectDao.selectList(wrapper);
                //将所需的信息按照文件名头分组，集中处理
                Map<Double, List<TbBXFileCollect>> mapValue = collects.stream().collect(Collectors.groupingBy(TbBXFileCollect::getFile_num));
                //k文件名头，v内容(将row里面的东西填满)
                mapValue.forEach((k, v) -> {
                    //文件编号
                    int i = k.intValue();
                    //文件路径
                    String directory = key.getString("path");
                    //文件名
                    String downloadFile = i + "-" + tbBxFileSave.getCoil_id() + ".json.tar";
                    //解析文件路径
                    ChannelSftp sftp = sFtpUtils.sftpConnect(host642);
                    if (directory != null && !"".equals(directory)) {
                        try {
                            sftp.cd(directory);
                        } catch (SftpException e) {
                            throw new RuntimeException(e);
                        }
                    }
                    //处理文件流
                    InputStream inputStream = null;
                    try {
                        inputStream = sftp.get(directory + downloadFile);
                    } catch (SftpException e) {
                        return;
                        //throw new RuntimeException(e);
                    }

                    //判断文件大小（需要保证自己电脑有D://TEMP/目录，没有可以修改）
                    sFtpUtils.loadFile(inputStream, "D://TEMP/" + downloadFile);
                    File file = new File("D://TEMP/" + downloadFile);
                    if (file.length() > 1024 * 1024) {
                        log.error(directory + downloadFile + "大文件处理文件大小异常---" + host642);
                        try {
                            inputStream.close();
                        } catch (IOException e) {
                            throw new RuntimeException(e);
                        }
                        file.delete();
                        sftp.disconnect();
                        try {
                            sftp.getSession().disconnect();
                        } catch (JSchException e) {
                            throw new RuntimeException(e);
                        }
                        //如果小文件也是很大
                        tbBxFileSave.setStatusftp(8.0);
                        UpdateWrapper<TbBxFileSave> updateWrapper = new UpdateWrapper<>();
                        updateWrapper.lambda().eq(TbBxFileSave::getCoil_id, tbBxFileSave.getCoil_id());
                        tbBxFileSaveDao.update(tbBxFileSave, updateWrapper);
                        return;
                    }
                    file.delete();
                    //读完再读一次
                    InputStream inputStream1 = null;
                    try {
                        inputStream1 = sftp.get(directory + downloadFile);
                    } catch (SftpException e) {
                        throw new RuntimeException(e);
                    }

                    if (inputStream1 != null) {
                        try {
                            //开始处理大文件中的小文件
                            GzipCompressorInputStream gzip = null;
                            gzip = new GzipCompressorInputStream(inputStream);
                            String s = sFtpUtils.readFileToString(gzip);
                            if (StringUtils.hasText(s)) {
                                //解析出宝信原始数据
                                JDataBigToSmall jDataBigToSmall = JSON.parseObject(s, JDataBigToSmall.class);
                                List<List<String>> row = jDataBigToSmall.getRow();
                                //循环获取里面的数据
                                for (TbBXFileCollect tbBXFileCollect : v) {
                                    int i1 = tbBXFileCollect.getRow_id().intValue();
                                    //根据指定的下标获取需要的值
                                    List<String> value = row.get(i1);
                                    List<Double> collect = value.stream().map(Double::parseDouble).collect(Collectors.toList());
                                    //存入别名，加内容
                                    map.put(tbBXFileCollect.getL1_variable_alias(), collect);
                                }
                            }
                            gzip.close();
                            inputStream1.close();
                            sftp.disconnect();
                            sftp.getSession().disconnect();
                        } catch (JSchException e) {
                            throw new RuntimeException(e);
                        } catch (IOException e) {
                            throw new RuntimeException(e);
                        }
                    }

                });
                //存入MongoDB
                jDataSave642.setRow(map);
                jDataSaveService642.insertData(jDataSave642);
                //如果处理完之后
                tbBxFileSave.setStatusftp(9.0);
                UpdateWrapper<TbBxFileSave> updateWrapper = new UpdateWrapper<>();
                updateWrapper.lambda().eq(TbBxFileSave::getCoil_id, tbBxFileSave.getCoil_id());
                tbBxFileSaveDao.update(tbBxFileSave, updateWrapper);
                JSONObject jsonObject = new JSONObject();
                jsonObject.put("volumeNo", key.getString("extid"));
                jsonObject.put("line", "C642");
                String jsonString = jsonObject.toJSONString();
                restTemplate.postForObject("http://170.0.200.112:1234/send_msg_to_models", jsonString, String.class);
                Thread.sleep(3000);
            }

        }


    }


}
